package org.mq.mymq.nameserver.store;

import org.mq.mymq.nameserver.common.CommonCache;
import org.mq.mymq.nameserver.common.TraceReplicationProperties;
import org.mq.mymq.nameserver.enums.ReplicationModeEnum;
import org.mq.mymq.nameserver.enums.ReplicationRoleEnum;
import org.mq.mymq.nameserver.event.model.ReplicationMsgEvent;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @ClassName ReplicationMsgQueueManager
 * @Description
 * @Author jiarong_ye
 * @Date 2025/6/23 10:42
 * @Version 1.0
 */
public class ReplicationMsgQueueManager {
    private BlockingQueue<ReplicationMsgEvent> replicationMsgQueue = new LinkedBlockingQueue<ReplicationMsgEvent>(5000);

    public void put(ReplicationMsgEvent replicationMsgEvent) {
        ReplicationModeEnum replicationModeEnum = ReplicationModeEnum.of(CommonCache.getNameserverProperties().getReplicationMode());
        ReplicationRoleEnum replicationRoleEnum = ReplicationRoleEnum.of(CommonCache.getNameserverProperties().getMasterSlaveReplicationProperties().getRole());
        // 当主从架构为主从复制且当前节点为主节点时，往阻塞队列添加代同步数据
        if (replicationModeEnum == ReplicationModeEnum.MASTER_SLAVE && replicationRoleEnum == ReplicationRoleEnum.MASTER) {
            this.sendMsgToQueue(replicationMsgEvent);
        } else if (replicationModeEnum == ReplicationModeEnum.TRACE) {
            TraceReplicationProperties traceReplicationProperties = CommonCache.getNameserverProperties().getTraceReplicationProperties();
            if (traceReplicationProperties.getNextNode() != null) {
                this.sendMsgToQueue(replicationMsgEvent);
            }
        }
    }

    private void sendMsgToQueue(ReplicationMsgEvent replicationMsgEvent) {
        try {
            replicationMsgQueue.put(replicationMsgEvent);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    public BlockingQueue<ReplicationMsgEvent> getReplicationMsgQueue() {
        return replicationMsgQueue;
    }
}
